// $Id: jpeg.c 111 2011-11-19 06:33:56Z cedric.shih@gmail.com $
/*
 * Copyright 2011 Cedric Shih (cedric dot shih at gmail dot com)
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/tcp.h>

#include <zcbuffer.h>

#include "log.h"

#include "jpeg.h"

#define DEFAULT_MAX_QTABLES 4
#define MIN(a,b) a == 0 ? b : (a > b ? b : a)
#define MAX(a,b) a > b ? a : b

int niortp_jpeg_parse(struct niortp_jpeg *transport, struct zcbuffer *frame,
		struct niortp_jpeg_size *size);

int niortp_jpeg_init(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame);

int niortp_jpeg_make_header(struct niortp_jpeg *transport);

int niortp_jpeg_sendto(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame);

int niortp_jpeg_start(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame);

int niortp_jpeg_write_header(struct niortp_jpeg *transport);

int niortp_jpeg_write_qtable(struct niortp_jpeg *transport);

int niortp_jpeg_write_payload(struct niortp_jpeg *transport);

ssize_t niortp_jpeg_clear_buffers(struct zcqueue *queue);

int niortp_jpeg_qos(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame);

void niortp_jpeg_stat_cb(int fd, short event, void *arg);

struct niortp_transport niortp_jpeg = {
		niortp_jpeg_create,
		niortp_jpeg_destroy,
		niortp_jpeg_setup,
		niortp_jpeg_teardown,
		niortp_jpeg_send,
		niortp_jpeg_pending,
		niortp_jpeg_next,
		niortp_jpeg_stat,
};

void *niortp_jpeg_create(struct event_base *evbase)
{
	int rc;
	struct niortp_jpeg *transport;

	if (!(transport = calloc(1, sizeof(*transport)))) {
		rc = errno ? errno : -1;
		niortsp_error("calloc: %s", strerror(rc));
		goto finally;
	}

	transport->evbase = evbase;

	rc = 0;

finally:
	if (rc != 0) {
		niortp_jpeg_destroy(transport);
		transport = NULL;
		errno = rc;
	}
	return transport;
}

void niortp_jpeg_destroy(void *obj)
{
	struct niortp_jpeg *transport = obj;

	if (!transport) {
		return;
	}

	if (transport->setup) {
		niortp_jpeg_teardown(transport);
	}

	free(transport);
}

int niortp_jpeg_setup(void *obj, int fd, struct niortp_transport_param *param)
{
	int rc;
	struct niortp_jpeg *transport = obj;

	if (!transport || !param) {
		rc = EINVAL;
		niortsp_error("null pointer(s)");
		goto finally;
	}

	if (transport->setup) {
		rc = EPERM;
		niortsp_error("already set up");
		goto finally;
	}

	niortsp_debug("setting up: %s:%d", inet_ntoa(param->remote.sin_addr),
			ntohs(param->remote.sin_port));

	switch(param->type) {
	case NIORTP_TRANSPORT_UDP_UNICAST:
		break;
	case NIORTP_TRANSPORT_INTERLEAVED:
		if (!(transport->queue = zcqueue_new())) {
			rc = errno ? errno : -1;
			niortsp_error("zcqueue_new: %s", strerror(rc));
			goto finally;
		}

		if (!(transport->dropping = zcqueue_new())) {
			rc = errno ? errno : -1;
			niortsp_error("zcqueue_new: %s", strerror(rc));
			goto finally;
		}
		break;
	default:
		rc = ENOSYS;
		niortsp_error("transport not supported: %d", param->type);
		break;
	}

	transport->max_qtables = DEFAULT_MAX_QTABLES;

	if (!(transport->qtables = malloc(transport->max_qtables *
			sizeof(*transport->qtables)))) {
		rc = errno ? errno : -1;
		niortsp_error("malloc: %s", strerror(rc));
		goto finally;
	}

	evtimer_set(&transport->ev, niortp_jpeg_stat_cb, transport);

	if (event_base_set(transport->evbase, &transport->ev)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to set event base: %s", strerror(rc));
		goto finally;
	}

	transport->tv.tv_sec = 1;
	transport->tv.tv_usec = 0;

	if (event_add(&transport->ev, &transport->tv)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to add timer: %s", strerror(rc));
		goto finally;
	}

	transport->fd = fd;
	memcpy(&transport->param, param, sizeof(transport->param));
	transport->setup = 1;

	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_teardown(void *obj)
{
	int rc;
	ssize_t n;
	struct niortp_jpeg *transport = obj;

	if (!transport) {
		rc = EINVAL;
		niortsp_error("null pointer(s)");
		goto finally;
	}

	if (!transport->setup) {
		rc = EPERM;
		niortsp_error("not yet set up");
		goto finally;
	}

	if (event_initialized(&transport->ev) &&
			evtimer_pending(&transport->ev, NULL)) {
		if ((rc = evtimer_del(&transport->ev))) {
			niortsp_error("failed to delete timer: %s",
					strerror(rc));
			goto finally;
		}
	}

	if (transport->buffer) {
		zcbuffer_free(transport->buffer);
		transport->buffer = NULL;
	}

	if (transport->dropping) {
		if ((n = niortp_jpeg_clear_buffers(
				transport->dropping)) < 0) {
			rc = errno ? errno : -1;
			niortsp_error("failed to clear buffers: %s",
					strerror(rc));
			goto finally;
		}

		zcqueue_free(transport->dropping);
		transport->dropping = NULL;
	}

	if (transport->queue) {
		if ((n = niortp_jpeg_clear_buffers(
				transport->queue)) < 0) {
			rc = errno ? errno : -1;
			niortsp_error("failed to clear buffers: %s",
					strerror(rc));
			goto finally;
		}

		zcqueue_free(transport->queue);
		transport->queue = NULL;
	}

	if (transport->qtables) {
		free(transport->qtables);
		transport->qtables = NULL;
	}

	transport->setup = 0;
	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_stat(void *obj, struct niortp_transport_stat *stat)
{
	int rc;
	struct niortp_jpeg *transport = obj;

	if (!transport || !stat) {
		rc = EINVAL;
		niortsp_error("null pointer(s)");
		goto finally;
	}

	if (!transport->setup) {
		rc = EPERM;
		niortsp_error("not yet set up");
		goto finally;
	}

	memcpy(stat, &transport->old, sizeof(*stat));
	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_send(void *obj, const struct timeval *timestamp,
		struct zcbuffer *frame)
{
	int rc;
	struct niortp_jpeg *transport = obj;

	if (!transport || !timestamp || !frame) {
		rc = EINVAL;
		niortsp_error("null pointer(s)");
		goto finally;
	}

	if (!transport->setup) {
		rc = EPERM;
		niortsp_error("not yet set up");
		goto finally;
	}

	transport->new.incoming.units++;
	transport->new.incoming.bytes += zcbuffer_remaining(frame);

	niortsp_trace("sending frame %p to: %s:%d", frame,
			inet_ntoa(transport->param.remote.sin_addr),
			ntohs(transport->param.remote.sin_port));

	if ((rc = niortp_jpeg_qos(transport, timestamp, frame))) {
		niortsp_error("failed to apply QoS rules: %s", strerror(rc));
		goto finally;
	}

	switch(transport->param.type) {
	case NIORTP_TRANSPORT_UDP_UNICAST:
		if ((rc = niortp_jpeg_sendto(transport,
				timestamp, frame))) {
			niortsp_error("failed to unicast: %s",
					strerror(rc));
			goto finally;
		}
		break;
	case NIORTP_TRANSPORT_INTERLEAVED:
		if ((rc = niortp_jpeg_start(transport,
				timestamp, frame))) {
			niortsp_error("failed to send frame: %s",
					strerror(rc));
			goto finally;
		}
		break;
	default:
		rc = ENOSYS;
		niortsp_error("transport not supported: %d",
				transport->param.type);
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_pending(void *obj)
{
	int pending;
	struct niortp_jpeg *transport = obj;

	if (!transport) {
		niortsp_error("null pointer");
		return 0;
	}

	if ((pending = (transport->param.type ==
			NIORTP_TRANSPORT_INTERLEAVED) &&
			(transport->buffer || (transport->queue &&
			zcqueue_size(transport->queue) > 0)))) {
		niortsp_trace("pending");
	} else {
		niortsp_trace("not pending");
	}

	return pending;
}

int niortp_jpeg_next(void *obj)
{
	int rc;
	struct niortp_jpeg *transport = obj;
	struct zcbuffer *buffer = NULL;
	struct timeval timestamp;

	if (!transport) {
		rc = EINVAL;
		niortsp_error("null pointer");
		goto finally;
	}

	if (transport->param.type != NIORTP_TRANSPORT_INTERLEAVED) {
		rc = EPERM;
		niortsp_error("not interleaved");
		goto finally;
	}

	if (!transport->buffer) {
		niortsp_trace("processing next frame");

		if (!(buffer = zcqueue_take(transport->queue,
				&timestamp))) {
			rc = errno ? errno : -1;
			niortsp_error("failed to take: %s", strerror(rc));
			goto finally;
		}

		if ((rc = niortp_jpeg_init(transport, &timestamp,
				buffer))) {
			niortsp_error("failed to init: %s", strerror(rc));
			goto finally;
		}

		if ((rc = niortp_jpeg_make_header(transport))) {
			niortsp_error("failed to make header: %s",
					strerror(rc));
			goto finally;
		}
	}

	niortsp_trace("continuing JPEG frame: %p", transport->buffer);

	if (transport->packet.header.written < transport->packet.header.len &&
			(rc = niortp_jpeg_write_header(transport))) {
		if (rc != EAGAIN && rc != EINPROGRESS) {
			niortsp_error("failed to write header: %s",
					strerror(rc));
		}
		goto finally;
	}

	while (transport->offset == 0 && transport->packet.qtable.index <
			transport->num_qtables) {
		if ((rc = niortp_jpeg_write_qtable(transport))) {
			if (rc != EAGAIN && rc != EINPROGRESS) {
				niortsp_error("failed to write qtable %d: %s",
						transport->packet.qtable.index,
						strerror(rc));
			}
			goto finally;
		}
	}

	if ((rc = niortp_jpeg_write_payload(transport))) {
		if (rc != EAGAIN && rc != EINPROGRESS) {
			niortsp_error("failed to write payload: %s",
					strerror(rc));
		}
		goto finally;
	}

	rc = 0;

finally:
	if (rc != 0) {
		if (buffer) {
			zcbuffer_free(buffer);
			transport->buffer = NULL;
		}
	}
	return rc;
}

int niortp_jpeg_parse(struct niortp_jpeg *transport, struct zcbuffer *frame,
		struct niortp_jpeg_size *size)
{
	int rc, b, i;
	unsigned char marker[2];
	u_int16_t len;
	size_t position;
	int data_precision, num_components;

	niortsp_trace("parsing JPEG frame: %p", frame);

	position = zcbuffer_position(frame);
	transport->num_qtables = 0;
	memset(size, '\0', sizeof(*size));

	if ((rc = zcbuffer_gets(frame, marker, sizeof(marker)))) {
		niortsp_error("failed to read marker: %s", strerror(rc));
		goto finally;
	}

	if (marker[0] != 0xff || marker[1] != 0xd8) {
		rc = EPROTO;
		niortsp_error("missing start of image");
		goto finally;
	}

	while (zcbuffer_remaining(frame) > 0) {
		if ((rc = zcbuffer_get(frame, &b))) {
			niortsp_error("failed to get: %s", strerror(rc));
			goto finally;
		}

		if (b != 0xff) {
			continue;
		}

		if ((rc = zcbuffer_peek(frame, zcbuffer_position(frame),
				&b))) {
			niortsp_error("failed to peek: %s", strerror(rc));
			goto finally;
		}

		if (b == 0xc0) {
			niortsp_trace("found start of frame: %x", b);

			if ((rc = zcbuffer_skip(frame, 1))) {
				niortsp_error("failed to skip: %s", strerror(rc));
				goto finally;
			}

			if ((rc = zcbuffer_get_u16(frame, &len))) {
				niortsp_error("failed to read length: %s",
						strerror(rc));
				goto finally;
			}

			// TODO: check length

			if ((rc = zcbuffer_get(frame, &data_precision))) {
				niortsp_error("failed to read data precision: %s",
						strerror(rc));
				goto finally;
			}

			if ((rc = zcbuffer_get_u16(frame, &len))) {
				niortsp_error("failed to read height: %s",
						strerror(rc));
				goto finally;
			}

			size->height = len;

			if ((rc = zcbuffer_get_u16(frame, &len))) {
				niortsp_error("failed to read width: %s",
						strerror(rc));
				goto finally;
			}

			size->width = len;

			if ((rc = zcbuffer_get(frame, &num_components))) {
				niortsp_error("failed to read number of "
						"components: %s", strerror(rc));
				goto finally;
			}

			niortsp_trace("precision=%d, size=(%d,%d), "
					"number of components=%d",
					data_precision, size->width,
					size->height, num_components);

		} else if (b == 0xdb) {
			i = transport->num_qtables;

			if ((rc = zcbuffer_skip(frame, 1))) {
				niortsp_error("failed to skip: %s", strerror(rc));
				goto finally;
			}

			niortsp_trace("found qtable %d at %d",
					transport->num_qtables,
					zcbuffer_position(frame));

			if (i >= transport->max_qtables) {
				rc = EPROTO;
				niortsp_error("more qtables than expected: "
						"%d > %d", i + 1,
						transport->max_qtables);
				goto finally;
			}

			if ((rc = zcbuffer_get_u16(frame, &len))) {
				niortsp_error("failed to read length: %s",
						strerror(rc));
				goto finally;
			}

			if ((rc = zcbuffer_skip(frame, 1))) {
				niortsp_error("failed to skip qtable precision: "
						"%s", strerror(rc));
				goto finally;
			}

			niortsp_trace("marking qtable %d at %d", i,
					zcbuffer_position(frame));

			transport->qtables[i].offset =
					zcbuffer_position(frame);
			transport->qtables[i].len = len - sizeof(len) - 1;

			if ((rc = zcbuffer_skip(frame,
					transport->qtables[i].len))) {
				niortsp_error("failed to skip qtable '%d': %s",
						transport->qtables[i].len,
						strerror(rc));
				goto finally;
			}

			transport->num_qtables++;
			continue;
		}

		if (b != 0xda) {
			continue;
		}

		if ((rc = zcbuffer_skip(frame, 1))) {
			niortsp_error("failed to skip: %s", strerror(rc));
			goto finally;
		}

		niortsp_trace("found start of scan at %d",
				zcbuffer_position(frame));

		if ((rc = zcbuffer_get_u16(frame, &len))) {
			niortsp_error("failed to read length: %s",
					strerror(rc));
			goto finally;
		}

		if ((rc = zcbuffer_skip(frame, len - sizeof(len)))) {
			niortsp_error("failed to skip start of scan header "
					"'%d': %s", len, strerror(rc));
			goto finally;
		}

		niortsp_trace("marking scanlines at %d",
				zcbuffer_position(frame));

		zcbuffer_set_position(frame, zcbuffer_position(frame));
		break;
	}

	if (size->width == 0 || size->height == 0) {
		rc = EBADMSG;
		niortsp_error("failed to find width or height");
		goto finally;
	}

	if (transport->num_qtables == 0) {
		rc = EBADMSG;
		niortsp_error("failed to find qtables");
		goto finally;
	}

	if (zcbuffer_position(frame) == position) {
		rc = EPROTO;
		niortsp_error("failed to find scanlines");
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_init(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *buffer)
{
	int rc;
	size_t i;
	struct timeval now, diff;
	unsigned int duration;

	if ((rc = niortp_jpeg_parse(transport, buffer, &transport->size))) {
		niortsp_error("failed to parse: %s", strerror(rc));
		goto finally;
	}

	memset(&transport->header, '\0', sizeof(transport->header));

	transport->header.interleaved.magic = '$';
	transport->header.interleaved.stream = ntohs(
			transport->param.address.rtp.sin_port);

	transport->header.rtp.a = 0x80;
	transport->header.rtp.b = 0x1a;
        transport->header.rtp.ts = htonl(
        		niortp_timeval_to_timestamp(timestamp));
        transport->header.rtp.ssrc = (u_int32_t)transport;

	transport->header.jpeg.type = 1; // Type
	transport->header.jpeg.q = 255; // Q
	transport->header.jpeg.width = transport->size.width >> 3;
	transport->header.jpeg.height = transport->size.height >> 3;

	transport->header.qtable.mbz = 0;
	transport->header.qtable.precision = 0;

	transport->size_qtables = 0;

	for (i = 0; i < transport->num_qtables; i++) {
		transport->size_qtables += transport->qtables[i].len;
	}

	transport->header.qtable.length[0] = (transport->size_qtables >> 8 & 0xff) << 8;
	transport->header.qtable.length[1] = (transport->size_qtables & 0xff);

	if (memcmp(&transport->timestamp, timestamp,
			sizeof(transport->timestamp))) {
		if ((rc = gettimeofday(&now, NULL)) == -1) {
			rc = errno ? errno : -1;
			niortsp_error("failed to get time: %s", strerror(rc));
			goto finally;
		}

		if (transport->start.tv_sec != 0 &&
				transport->start.tv_usec != 0) {
			if ((rc = niortp_timeval_diff(&transport->start,
					&now, &diff))) {
				niortsp_error("failed to diff time: %s",
						strerror(rc));

				memset(&transport->start, '\0',
						sizeof(transport->start));
			} else {
				duration = diff.tv_sec * 1000000 + diff.tv_usec;
				transport->new.duration.min = MIN(
						transport->new.duration.min,
						duration);
				transport->new.duration.max = MAX(
						transport->new.duration.max,
						duration);
			}
		}

		memcpy(&transport->start, &now, sizeof(transport->start));
	}

	memcpy(&transport->timestamp, timestamp, sizeof(transport->timestamp));
	transport->buffer = buffer;
	transport->offset = 0;
	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_make_header(struct niortp_jpeg *transport)
{
	int rc, tcp, marker;
	size_t remaining, header, qtables, max, payload;

	niortsp_trace("making header: offset=%d", transport->offset);

	tcp = transport->param.type == NIORTP_TRANSPORT_INTERLEAVED;

	remaining = zcbuffer_remaining(transport->buffer);
	header = (tcp ? sizeof(transport->header.interleaved) : 0) +
			 sizeof(transport->header.rtp) +
			sizeof(transport->header.jpeg);
	qtables = 0;

	if (transport->offset == 0) {
		// need to write qtables for the first frame
		header += sizeof(transport->header.qtable);
		qtables += transport->size_qtables;
	}

	if ((max = transport->param.mtu - IP_HEADER_MAX -
			(tcp ? TCP_HEADER_MAX : UDP_HEADER_MAX) -
			header - qtables) < 0) {
		rc = EOVERFLOW;
		niortsp_error("insufficient MTU size: %d", transport->param.mtu);
		goto finally;
	}

	marker = max >= remaining;
	payload = remaining > max ? max : remaining;

	if (tcp) {
		transport->header.interleaved.len = htons(header -
				sizeof(transport->header.interleaved) +
				qtables + payload);
	}

        transport->header.rtp.seq = htons(transport->sequence++);
        transport->new.rtp.units++;

	if (marker) {
		niortsp_trace("turning on marker bit");
		transport->header.rtp.b |= 0x80; // Marker=1
	}

	transport->header.jpeg.tspec_off =
			0x00 << 24 | htonl(transport->offset);

	transport->packet.header.len = header;
	transport->packet.header.written = 0;
	transport->packet.qtable.index = 0;
	transport->packet.qtable.written = 0;
	transport->packet.len = payload;
	transport->packet.written = 0;
	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_sendto(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame)
{
	int rc, fd;
	unsigned int i;
	struct sockaddr_in *addr;
	unsigned char *data;
	size_t len;

	niortsp_trace("sending frame %p: %s:%d", frame,
			inet_ntoa(transport->param.remote.sin_addr),
			ntohs(transport->param.remote.sin_port));

	if ((rc = niortp_jpeg_init(transport, timestamp, frame))) {
		niortsp_error("failed to init: %s", strerror(rc));
		goto finally;
	}

	fd = transport->fd;
	addr = &transport->param.address.rtp;

	while ((zcbuffer_remaining(frame)) > 0) {
		if ((rc = niortp_jpeg_make_header(transport))) {
			niortsp_error("failed to make header: %s",
					strerror(rc));
			goto finally;
		}

		data = (unsigned char *)&transport->header.rtp;
		len = transport->packet.header.len;

		niortsp_trace("sending header: %d@%p", len, data);

		if (sendto(fd, data, len, MSG_DONTWAIT|MSG_MORE,
				(struct sockaddr *)addr,
				sizeof(*addr)) != len) {
			rc = errno ? errno : -1;
			niortsp_error("failed to unicast: %s", strerror(rc));
			goto finally;
		}

		transport->new.rtp.bytes += len;

		if (transport->offset == 0) {
			for (i = 0; i < transport->num_qtables; i++) {
				data = zcbuffer_array(frame) +
						transport->qtables[i].offset;
				len = transport->qtables[i].len;

				niortsp_trace("sending qtable %d: %d@%p",
						i, len, data);

				if (sendto(fd, data, len,  MSG_DONTWAIT|MSG_MORE,
						(struct sockaddr *)addr,
						sizeof(*addr)) != len) {
					rc = errno ? errno : -1;
					niortsp_error("failed to write "
							"qtable %d: %s", i,
							strerror(rc));
					goto finally;
				}

				transport->new.rtp.bytes += len;
			}
		}

		data = zcbuffer_array(frame) + zcbuffer_position(frame);
		len = transport->packet.len;

		niortsp_trace("sending data: %d@%p", len, data);

		if (sendto(fd, data, len, MSG_DONTWAIT,
				(struct sockaddr *)addr,
				sizeof(*addr)) != len) {
			rc = errno ? errno : -1;
			niortsp_error("failed to write data: %s", strerror(rc));
			goto finally;
		}

		transport->new.rtp.bytes += len;

		if ((rc = zcbuffer_skip(frame, len))) {
			niortsp_error("failed to skip: %s", strerror(rc));
			goto finally;
		}

		transport->offset += len;
	}

	if (memcmp(&transport->previous, &transport->timestamp,
			sizeof(transport->previous))) {
		transport->new.instances++;
		memcpy(&transport->previous, &transport->timestamp,
				sizeof(transport->previous));
	}

	transport->buffer = NULL;
	transport->new.outgoing.units++;
	transport->new.outgoing.bytes += transport->offset;

	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_qos(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame)
{
	int rc;
	ssize_t n;
	size_t dropped, queued;

	if (transport->param.type == NIORTP_TRANSPORT_UDP_UNICAST) {
		niortsp_trace("no need for apply Qos");
		rc = 0;
		goto finally;
	}
	
	dropped = 0;

	if (transport->buffer) {
		niortsp_trace("clearing frame(s) newer than %ld.%06ld",
				transport->timestamp.tv_sec,
				transport->timestamp.tv_usec);

		if ((rc = zcqueue_take_newer(transport->queue,
				transport->dropping, &transport->timestamp))) {
			niortsp_error("failed to take newer: %s", strerror(rc));
			goto finally;
		}

		dropped = zcqueue_size(transport->dropping);

		if ((n = niortp_jpeg_clear_buffers(
				transport->dropping)) < 0) {
			rc = errno ? errno : -1;
			niortsp_error("failed to clear: %s", strerror(rc));
			goto finally;
		}

		transport->new.dropped.units += dropped;
		transport->new.dropped.bytes += n;
	}

	if (dropped > 0) {
		queued = zcqueue_size(transport->queue);
		
		niortsp_warn("qos applied %s:%d => dropped=%d, queued=%d",
				inet_ntoa(transport->param.remote.sin_addr),
				ntohs(transport->param.remote.sin_port),
				dropped, queued);
	}

	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_start(struct niortp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame)
{
	int rc;
	struct zcbuffer *clone = NULL;

	niortsp_trace("streaming frame %p: %s:%d", frame,
			inet_ntoa(transport->param.remote.sin_addr),
			ntohs(transport->param.remote.sin_port));

	if ((rc = niortp_jpeg_qos(transport, timestamp, frame))) {
		niortsp_error("failed to apply QoS rules: %s", strerror(rc));
		goto finally;
	}

	if (!(clone = zcbuffer_clone(frame))) {
		rc = errno ? errno : -1;
		niortsp_error("failed to clone: %s",
				strerror(rc));
		goto finally;
	}

	if ((rc = zcqueue_put(transport->queue, timestamp, clone))) {
		niortsp_error("failed to put: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	if (rc != 0) {
		if (clone) {
			zcbuffer_free(clone);
		}
	}
	return rc;
}

int niortp_jpeg_write_header(struct niortp_jpeg *transport)
{
	int rc;
	ssize_t n;
	void *data;
	size_t len;

	niortsp_trace("writing header: %d/%d bytes",
			transport->packet.header.written,
			transport->packet.header.len);

	data = &transport->header + transport->packet.header.written;
	len = transport->packet.header.len - transport->packet.header.written;

	if ((n = write(transport->fd, data, len)) <= 0) {
		rc = errno ? errno : -1;

		if (rc != EAGAIN && rc != EINPROGRESS) {
			niortsp_error("failed to write: %s", strerror(rc));
		}

		goto finally;
	}

	niortsp_trace("header written: %d bytes", n);

	transport->packet.header.written += n;
	transport->new.rtp.bytes += n;
	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_write_qtable(struct niortp_jpeg *transport)
{
	int rc;
	struct niortp_slice *qtable;
	unsigned char *data;
	size_t len;
	ssize_t n;

	if (transport->offset != 0) {
		rc = EPERM;
		niortsp_error("attempting to write qtable while offset != 0");
		goto finally;
	}

	if (transport->packet.qtable.index >= transport->num_qtables) {
		rc = EPERM;
		niortsp_error("index out of bound: %d >= %d",
				transport->packet.qtable.index,
				transport->num_qtables);
		goto finally;
	}

	qtable = transport->qtables + transport->packet.qtable.index;

	if (transport->packet.qtable.written >= qtable->len) {
		rc = ENODATA;
		niortsp_error("all data already written: %d >= %d",
				transport->packet.qtable.written, qtable->len);
		goto finally;
	}

	niortsp_trace("writing qtable %d: %d/%d bytes",
			transport->packet.qtable.index,
			transport->packet.qtable.written, qtable->len);

	data = zcbuffer_array(transport->buffer) + qtable->offset +
			transport->packet.qtable.written;
	len = qtable->len - transport->packet.qtable.written;

	if ((n = write(transport->fd, data, len)) <= 0) {
		rc = errno ? errno : -1;

		if (rc != EAGAIN && rc != EINPROGRESS) {
			niortsp_error("failed to write: %s", strerror(rc));
		}

		goto finally;
	}

	niortsp_trace("qtable %d written: %d bytes",
			transport->packet.qtable.index, n);

	transport->packet.qtable.written += n;
	transport->new.rtp.bytes += n;

	if (transport->packet.qtable.written >= qtable->len) {
		niortsp_trace("qtable %d transport completed",
				transport->packet.qtable.index);

		transport->packet.qtable.index++;
		transport->packet.qtable.written = 0;
	}

	rc = 0;

finally:
	return rc;
}

int niortp_jpeg_write_payload(struct niortp_jpeg *transport)
{
	int rc;
	ssize_t n;
	unsigned char *data;
	size_t len;

	if (zcbuffer_remaining(transport->buffer) <= 0) {
		rc = ENODATA;
		niortsp_error("buffer already written");
		goto finally;
	}

	niortsp_trace("writing payload: %d/%d bytes",
			transport->packet.written, transport->packet.len);

	data = zcbuffer_array(transport->buffer) +
			zcbuffer_position(transport->buffer) +
			transport->packet.written;
	len = transport->packet.len - transport->packet.written;

	if ((n = write(transport->fd, data, len)) <= 0) {
		rc = errno ? errno : -1;

		if (rc != EAGAIN && rc != EINPROGRESS) {
			niortsp_error("failed to write: %s", strerror(rc));
		}

		goto finally;
	}

	niortsp_trace("payload written: %d bytes", n);

	transport->packet.written += n;
	transport->new.rtp.bytes += n;

	if (transport->packet.written >= transport->packet.len) {
		niortsp_trace("packet transport completed; incrementing "
				"position: %d => %d",
				zcbuffer_position(transport->buffer),
				zcbuffer_position(transport->buffer) +
				transport->packet.len);

		if ((rc = zcbuffer_skip(transport->buffer,
				transport->packet.len))) {
			niortsp_error("failed to skip: %s", strerror(rc));
			goto finally;
		}

		transport->offset += transport->packet.len;

		if (zcbuffer_remaining(transport->buffer) > 0) {
			niortsp_trace("preparing next packet");

			if ((rc = niortp_jpeg_make_header(transport))) {
				niortsp_error("failed to make header: %s",
						strerror(rc));
				goto finally;
			}
		} else {
			niortsp_trace("frame transport completed: %p",
					transport->buffer);

			zcbuffer_free(transport->buffer);
			transport->buffer = NULL;
			transport->new.outgoing.units++;
			transport->new.outgoing.bytes += transport->offset;

			if (memcmp(&transport->previous,
					&transport->timestamp,
					sizeof(transport->previous))) {
				transport->new.instances++;
				memcpy(&transport->previous,
						&transport->timestamp,
						sizeof(transport->previous));
			}
		}
	}

	rc = 0;

finally:
	return rc;
}

ssize_t niortp_jpeg_clear_buffers(struct zcqueue *queue)
{
	int rc;
	struct zcbuffer *buffer;
	ssize_t size = 0;

	while (zcqueue_size(queue) > 0) {
		if (!(buffer = zcqueue_take(queue, NULL))) {
			rc = errno ? errno : -1;
			niortsp_error("failed to take: %s", strerror(rc));
			goto finally;
		}
		size += zcbuffer_remaining(buffer);
		zcbuffer_free(buffer);
	}

	rc = 0;

finally:
	if (rc != 0) {
		size = -1;
		errno = rc;
	}
	return size;
}

void niortp_jpeg_stat_cb(int fd, short event, void *arg)
{
	struct niortp_jpeg *transport = arg;

	memcpy(&transport->old, &transport->new, sizeof(transport->old));
	memset(&transport->new, '\0', sizeof(transport->new));
	event_add(&transport->ev, &transport->tv);
}
